package com.hexinfo.dmpro.dev.data.transfer.handle.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.hexinfo.dmpro.data.transfer.common.common.manager.AgentManager;
import com.hexinfo.dmpro.dev.data.transfer.common.enums.DevDataTransferImportTaskStatusEnum;
import com.hexinfo.dmpro.dev.data.transfer.common.enums.DevDataTransferListStatusEnum;
import com.hexinfo.dmpro.dev.data.transfer.handle.IDevDataImportHandler;
import com.hexinfo.dmpro.dev.data.transfer.model.DevDataImportList;
import com.hexinfo.dmpro.dev.data.transfer.model.DevDataImportTask;
import com.hexinfo.dmpro.dev.data.transfer.service.IDevDataImportListService;
import com.hexinfo.dmpro.dev.data.transfer.service.IDevDataImportTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.Arrays;

@Slf4j
@Service
public class DevDataImportHandler implements IDevDataImportHandler {

    @Autowired
    private IDevDataImportListService devDataImportListService;

    @Autowired
    private IDevDataImportTaskService devDataImportTaskService;

    @Autowired
    private AgentManager agentManager;

    @Value("${conn.pool.hive.target.url}")
    private String hiveJdbcUrl;

    @Value("${conn.pool.hive.target.username}")
    private String hiveUserName;

    @Value("${conn.pool.hive.target.password}")
    private String hivePassword;

    @Value("${dev-data-ftp.server}")
    private String ftpServer;

    @Value("${dev-data-ftp.port}")
    private Integer ftpPort;

    @Value("${dev-data-ftp.user}")
    private String ftpUserName;

    @Value("${dev-data-ftp.password}")
    private String ftpPassword;

    @Value("${nameNode.dev.nameNodeIp}")
    private String hdfsNameNodeHosts;

    @Value("${impala-shell.address}")
    private String impalaShellAddress;

    @Value("${impala-shell.username}")
    private String impalaShellUserName;

    @Value("${impala-shell.password}")
    private String impalaShellPassword;

    /*
     * @Method: execute <br>
     * @Param: [devDataImportList] <br>
     * @Return: void <br>
     * @Description：轮询分发开发数据导入任务执行<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/19 14:32 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    @Async("devDataTransferAsyncTaskExecutor")
    public void execute(DevDataImportList devDataImportList) {
        try {
            // 判断任务状态是否中止
            if (importTaskIsAbolish(devDataImportList.getTaskCode())) {
                // 修改清单状态
                abolishImportList(devDataImportList);
            }else if (importTaskIsRun(devDataImportList.getTaskCode())) {
                // ① 补充导入清单配置信息
                supplementDevDataExportList(devDataImportList);
                // ② 执行开发数据同步任务
                executeTask(devDataImportList);
            } else {
                log.warn(StrUtil.format("表【{}】开发数据导入暂停执行", devDataImportList.getDbAndTableName()));
                ThreadUtil.sleep(1000);
                devDataImportListService.restart(devDataImportList.getId());
            }
        } catch (Exception e) {
            log.error(StrUtil.format("表【{}】开发数据导出异常, 异常信息: {}", devDataImportList.getDbAndTableName(), e.getMessage()), e);
            updateDevDataImportListToFail(devDataImportList, e.getMessage());
        }
    }

    /**
     * @Method importTaskIsAbolish
     * @Param
     * @param taskCode
     * @Return boolean
     * @Description 判断任务是否中止执行
     * @Author Wang zhihao
     * @Date 2024/1/3 19:06
     * @Version V1.0
     */
    private boolean importTaskIsAbolish(String taskCode) {
        DevDataImportTask devDataImportTask = devDataImportTaskService.queryByTaskCode(taskCode);
        return DevDataTransferImportTaskStatusEnum.abolish.getValue().equals(devDataImportTask.getTaskStatus());
    }

    /**
     * @Method abolishImportList
     * @Param
     * @param devDataImportList
     * @Return void
     * @Description 修改清单状态
     * @Author Wang zhihao
     * @Date 2024/1/3 20:40
     * @Version V1.0
     */
    private void abolishImportList(DevDataImportList devDataImportList) {
        if (DevDataTransferListStatusEnum.TODO.getValue().equals(devDataImportList.getImportStatus())) {
            devDataImportList.setImportStatus(DevDataTransferListStatusEnum.ABOLISH.getValue());
            devDataImportListService.updateById(devDataImportList);
        }
    }

    private boolean importTaskIsRun(String taskCode) {
        DevDataImportTask devDataImportTask = devDataImportTaskService.queryByTaskCode(taskCode);
        return !DevDataTransferImportTaskStatusEnum.suspend.getValue().equals(devDataImportTask.getTaskStatus());
    }

    /*
     * @Method: executeTask <br>
     * @Param: [devDataImportList] <br>
     * @Return: void <br>
     * @Description：发送给可用AGENT执行开发数据导入任务，若异常则任务回到队列中<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/19 14:32 <br>
     * @Version： V2.0.2<br>
     */
    private synchronized void executeTask(DevDataImportList devDataImportList) {
        try {
            // ③ 获取可用Agent
            String agentKey = getAvailableAgent();
            devDataImportList.setAgentKey(agentKey);
            // ④ 执行开发数据同步任务
            sendAgentExport(devDataImportList, agentKey);
            // ⑤ 更新导入任务清单状态为进行中
            updateDevDataImportListToRunning(devDataImportList, agentKey);
            // ⑥ AGNET新增执行任务
            agentManager.addExecuteTask(agentKey);
        } catch (Exception e) {
            log.warn(StrUtil.format("表【{}】开发数据导入发送执行失败: {}", devDataImportList.getDbAndTableName(), e.getMessage()));
            ThreadUtil.sleep(30000);
            // 发送异常则重新放回队列中等待二次处理
            devDataImportListService.restart(devDataImportList.getId());
        }
    }

    /*
     * @Method: updateDevDataImportListToRunning <br>
     * @Param: [devDataImportList] <br>
     * @Return: void 更新导入任务为执行中<br>
     * @Description：<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/19 14:32 <br>
     * @Version： V2.0.2<br>
     */
    private void updateDevDataImportListToRunning(DevDataImportList devDataImportList, String agentKey) {
        devDataImportList.setStartTime(DateUtil.date());
        devDataImportList.setImportStatus(DevDataTransferListStatusEnum.RUNNING.getValue());
        devDataImportList.setAgentKey(agentKey);
        devDataImportListService.updateById(devDataImportList);
    }

    /*
     * @Method: getAvailableAgent <br>
     * @Param: [] <br>
     * @Return: com.hexinfo.dmpro.data.transfer.common.model.AgentInfo <br>
     * @Description：获取可用的数据迁移服务<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/6 19:04 <br>
     * @Version： V2.0.2<br>
     */
    private String getAvailableAgent() {
        String availableAgent = agentManager.getAvailableAgent();
        if (StrUtil.isBlank(availableAgent)) {
            throw new RuntimeException("没有可用的执行AGENT，请等待AGENT释放资源！");
        }
        return availableAgent;
    }

    /*
     * @Method: supplementDevDataExportList <br>
     * @Param: [devDataImportList] <br>
     * @Return: void <br>
     * @Description：补充执行导入任务需要的配置信息<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/19 14:34 <br>
     * @Version： V2.0.2<br>
     */
    private void supplementDevDataExportList(DevDataImportList devDataImportList) {
        devDataImportList.setHdfsTmpPath("/tmp/dmpro");
        devDataImportList.setFtpHost(ftpServer);
        devDataImportList.setFtpPort(ftpPort);
        devDataImportList.setFtpUserName(ftpUserName);
        devDataImportList.setFtpPassword(ftpPassword);
        devDataImportList.setHiveJdbcUrl(hiveJdbcUrl);
        devDataImportList.setHiveUserName(hiveUserName);
        devDataImportList.setHivePassword(hivePassword);
        devDataImportList.setHdfsNameNodeHosts(Arrays.asList(StrUtil.split(hdfsNameNodeHosts, ",")));
        devDataImportList.setHdfsPort(8020);
        devDataImportList.setImpalaSheelAddress(impalaShellAddress);
        devDataImportList.setImpalaUserName(impalaShellUserName);
        devDataImportList.setImpalaPassword(impalaShellPassword);
    }

    /*
     * @Method: sendAgentExport <br>
     * @Param: [devDataImportList, agentKey] <br>
     * @Return: void <br>
     * @Description：向AGENT发送开发数据导入任务<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/19 14:34 <br>
     * @Version： V2.0.2<br>
     */
    private void sendAgentExport(DevDataImportList devDataImportList, String agentKey) {
        try {
            String httpUrl = StrUtil.format("http://{}/api/hive/import", agentKey);
            final HttpRequest post = HttpUtil.createPost(httpUrl);
            String body = JSON.toJSONString(devDataImportList);
            log.info("表【{}】开发数据导入请求参数: {}", devDataImportList.getDbAndTableName(), body);
            final HttpResponse execute = post.body(body).execute();
            if (200 != execute.getStatus()) {
                throw new RuntimeException("HTTP请求失败: " + execute.body());
            }
            log.info("表【{}】开发数据导入请求成功！(●'◡'●)", devDataImportList.getDbAndTableName());
        } catch (Exception e) {
            throw new RuntimeException(StrUtil.format("发送HTTP请求失败, 失败原因{}", e.getMessage()), e);
        }
    }

    /*
     * @Method: updateDevDataImportListToFail <br>
     * @Param: [devDataImportList, errorMessage] <br>
     * @Return: void <br>
     * @Description：更新开发数据导入任务状态为失败<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/19 14:34 <br>
     * @Version： V2.0.2<br>
     */
    private void updateDevDataImportListToFail(DevDataImportList devDataImportList, String errorMessage) {
        devDataImportList.setEndTime(DateUtil.date());
        devDataImportList.setErrorMessage(errorMessage);
        devDataImportList.setImportStatus(DevDataTransferListStatusEnum.FAIL.getValue());
        devDataImportListService.updateById(devDataImportList);
        devDataImportTaskService.modifyFailNumByTaskCode(devDataImportList.getTaskCode());
    }
}
